package com.mutouren.modules.datapush.archives;

import java.io.File;
import java.util.Date;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Service;


import com.mutouren.common.log.LogAlias;
import com.mutouren.common.log.LogManager;
import com.mutouren.common.log.Logger;
import com.mutouren.common.thread.ThreadPoolManager;
import com.mutouren.common.utils.FileUtils;
import com.mutouren.common.utils.JsonUtils;
import com.mutouren.modules.datapush.base.DataPushLogAlias;
import com.mutouren.modules.datapush.base.PushMessage;
import com.mutouren.modules.datapush.base.PushStatus;
import com.mutouren.modules.datapush.dao.PushMessageDao;
import com.mutouren.modules.datapush.dao.impl.DaoFactoryImpl;
import com.mutouren.modules.datapush.other.ExceptionHelper;
import com.mutouren.modules.datapush.storage.StorageHelper;

@Deprecated
public class Reciever_v0 {
	
	//@Value("${datapush.reciever.maxPermitCacheSize:50000}")
	private int maxPermitSize_Reciever = 50000;
	//@Value("${datapush.reciever.maxThreadSize:3}")
	private int maxThreadSize_Reciever = 1;

	private Queue<PushMessage> messageQueue;
	
	private ExecutorService leader;
	private ExecutorService workers; 
	
	private Object waitTaskEvent = new Object();
	private Object waitWorkerEvent = new Object();
	private Object lockQueue = new Object();
	
	@Value("${datapush.reciever.storageFilePath}")
	private String storageFilePath;
	private volatile boolean isOpened = false;
	private volatile boolean isDataBaseException = false;
	
	//@Resource
	private PushMessageDao pushMessageDao = DaoFactoryImpl.getInstance().getPushMessageDao();
	
	protected final static Logger monitorLogger = LogManager.getLogger(DataPushLogAlias.DataPushMonitor.name());
	protected final static Logger errorLogger = LogManager.getLogger(LogAlias.SystemError.name());
	protected final static Logger runLogger = LogManager.getLogger(LogAlias.SystemRun.name());
	
	public Reciever_v0(int maxPermitSize) {
		messageQueue = new LinkedList<PushMessage>();
		this.maxPermitSize_Reciever = maxPermitSize;

		leader = ThreadPoolManager.createSingleThreadExecutor();
		workers = ThreadPoolManager.createThreadPool(0, maxThreadSize_Reciever, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
	}
	
	private void run() {
		this.isOpened = true;
		leader.execute(()-> this.dispatcherTask());	
	}
	
	private void stop() {
		this.isOpened = false;
		leader.shutdownNow();
		workers.shutdownNow();
		
		try {
			leader.awaitTermination(10, TimeUnit.SECONDS);
			workers.awaitTermination(10, TimeUnit.SECONDS);			
		} catch (Exception e) {
			errorLogger.error("reciever server awaitTermination error", e);
		}
	}	
	
	private void load() {
		synchronized(this.lockQueue) {	
			StorageHelper.load(storageFilePath, messageQueue);
			FileUtils.deleteFile(storageFilePath);
			runLogger.info("reciever server load message count: " + messageQueue.size());	
		}
	}
	
	private void save() {
		if (FileUtils.isExist(storageFilePath)) {
			FileUtils.deleteFile(storageFilePath);
		} else {
			FileUtils.createFolderIfNoExist(new File(storageFilePath).getParent());
		}
		
		synchronized(this.lockQueue) {	
			int saveCount = messageQueue.size();
			StorageHelper.save(storageFilePath, messageQueue);
				
			runLogger.info("reciever server save message count: " + saveCount);
		}		
	}
	
	public synchronized void open() {
		if (!this.isOpened) {
			load();
			run();
		}
	}
	
	public synchronized void close() {
		stop();
		save();
	}
	
	private long lastMoniterTime = 0;
	private void debugMoniter() {
		long curTime = new Date().getTime();		
		if ((curTime - lastMoniterTime) > 5000) {	
			String log = String.format("reciver server queue size:%d", messageQueue.size());
			monitorLogger.debug(log);
			lastMoniterTime = curTime;
		}
	}	
	
	private void dispatcherTask() {
		do {
			debugMoniter();
			PushMessage message = pollMessageFromQueue();
			processMessage(message);
		
			if ((getQueueSize() > 0)&&(this.isOpened)&&(!this.isDataBaseException)) continue;		
							
			synchronized(this.waitTaskEvent) {
				try {
					waitTaskEvent.wait(1000);
				} catch (InterruptedException e) {
					errorLogger.warn("reciever dispatcherTask Interrupted", e);
				}
			}
			
		} while(ThreadPoolManager.isRun()&&this.isOpened);		
	}
	
	private void processMessage(PushMessage message) {
		if (message == null) return;
				
		MyTask myTask = new MyTask(this.waitWorkerEvent, new Runnable() {
			@Override
			public void run() {
				try {
					saveToDatabase(message);					
				} catch(Throwable t) {
					errorLogger.error("recierve server.processMessage.run error", t);
				}
			}
		});	
		
		do {
			try {
				this.workers.submit(myTask);				
				return;
			} catch(RejectedExecutionException e) {
				
	            synchronized(this.waitWorkerEvent) {
	            	try {
						this.waitWorkerEvent.wait(1000);
					} catch (InterruptedException e1) {
						errorLogger.error("recierve server.processMessage wait error", e1);
					}
	            }				
			}
			
			// if close, refund message
			if (!this.isOpened) {
				synchronized(this.lockQueue) {
					messageQueue.add(message);
				}
			}
			
		} while(ThreadPoolManager.isRun()&&this.isOpened);
	}
	
	private void saveToDatabase(PushMessage message) {
		try {
			pushMessageDao.insert(message);
			isDataBaseException = false;
		} catch (Throwable t) {
			isDataBaseException = true;
			if (ExceptionHelper.isCommunicationsException(t)) {
				refundMessageToQueue(message);
			} else {
				errorLogger.error("no CommunicationsException, date discarded: " + JsonUtils.beanToJson(message), t);
			}
			errorLogger.error("recierve server.saveToDatabase error", t);
		}
	}	
	
	private int getQueueSize() {
		synchronized(this.lockQueue) {
			return messageQueue.size();
		}
	}
	
	private PushMessage pollMessageFromQueue() {
		synchronized(this.lockQueue) {			
			return messageQueue.poll();
		}
	}
	
	private void refundMessageToQueue(PushMessage message) {
		synchronized(this.lockQueue) {			
			messageQueue.add(message);
		}
	}
				
	public boolean receive(PushMessage message) {
		synchronized(this.lockQueue) {
			if (messageQueue.size() > maxPermitSize_Reciever) {
				return false;
			}
			
			message.setRequestCount(0);
			message.setPushStatus(PushStatus.INIT);
			message.setIsFinish(false);
			messageQueue.add(message);
		}
		synchronized(this.waitTaskEvent) {
			this.waitTaskEvent.notify();
		}		
		return true;
	}
	
    public static class MyTask implements Runnable {
        private Object lock;
        private Runnable task;

        public MyTask(Object lock, Runnable task) {
            this.lock = lock;
            this.task = task;
        }

        @Override
        public void run() {
            task.run();
            synchronized(lock) {
            	lock.notify();
            }
        }
    }		
		
}
